package com.luowenbo.shortlink.server.application.consumer;

import com.luowenbo.shortlink.server.application.consumer.request.TransformEvent;
import com.luowenbo.shortlink.server.infra.common.TimeZoneConstant;
import com.luowenbo.shortlink.server.infra.support.MappedDiagnosticContextProvider;
import com.luowenbo.shortlink.server.infra.util.BeanCopierUtils;
import com.luowenbo.shortlink.server.infra.util.JacksonUtils;
import com.luowenbo.shortlink.server.model.entity.TransformEventRecord;
import com.luowenbo.shortlink.server.service.TransformEventService;
import lombok.NoArgsConstructor;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;

import java.time.Instant;
import java.time.OffsetDateTime;

/**
 * @author luowenbo
 * @version v1
 * @description
 * @since 2022/08/28
 */
@RequiredArgsConstructor
@Component
@Slf4j
@RocketMQMessageListener(
        topic = "TransformEvent",                      // 1.topic：消息的发送者使用同一个topic
        consumerGroup = "test-group",               // 2.group：不用和生产者group相同 ( 在RocketMQ中消费者和发送者组没有关系 )
        selectorExpression = "*",                   // 3.tag：设置为 * 时，表示全部。
        messageModel = MessageModel.CLUSTERING    // 4.消费模式：默认 CLUSTERING （ CLUSTERING：负载均衡 ）（ BROADCASTING：广播机制 ）
)
public class TransformEventConsumer implements RocketMQListener<String> {

    private final TransformEventService transformEventService;
    private final MappedDiagnosticContextProvider mappedDiagnosticContextProvider;

    /**
     * 接收和处理URL转换事件
     */
    @Override
    public void onMessage(String content){
        mappedDiagnosticContextProvider.process(() -> {
            log.info("接收到URL转换事件,内容:{}......", content);
            TransformEvent event = JacksonUtils.X.parse(content, TransformEvent.class);
            TransformEventRecord record = new TransformEventRecord();
            BeanCopierUtils.X.copy(event, record);
            record.setRecordTime(OffsetDateTime.ofInstant(Instant.ofEpochMilli(event.getTimestamp()), TimeZoneConstant.CHINA.getZoneId()));
            record.setTransformStatus(event.getTransformStatusValue());
            record.setShortUrl(event.getShortUrlString());
            record.setLongUrl(event.getLongUrlString());
            transformEventService.recordTransformEvent(record);
            log.info("记录URL转换事件完成......");
        });
    }


}
